package com.xiaochong.tir.trades.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.tir.common.data.hbasepo.TradesHistory;
import com.xiaochong.tir.common.data.hbasepo.TradesHistoryCountFive;
import com.xiaochong.tir.common.data.hbasepo.TradesToKLine;
import com.xiaochong.tir.common.data.mapper.PurchaseOrderMapper;
import com.xiaochong.tir.common.data.mapper.TradeKlineMapper;
import com.xiaochong.tir.common.data.po.PurchaseOrder;
import com.xiaochong.tir.common.data.po.TradeKline;
import com.xiaochong.tir.common.data.redisPo.KActionTotalPo;
import com.xiaochong.tir.common.exception.ErrorDataException;
import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import com.xiaochong.tir.common.util.DateUtils;
import com.xiaochong.tir.common.util.HbaseTableUtils;
import com.xiaochong.tir.common.util.RedisKeyApp;
import com.xiaochong.tir.common.util.RedisKeysUtils;
import com.xiaochong.tir.trades.component.hbase.CurrencyInfoHbaseComponent;
import com.xiaochong.service.common.config.redis.RedisServer;
import com.xiaochong.tir.trades.rabbit.ExchangePairAllMqQueue;
import net.sf.json.util.JSONTokener;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.MathContext;
import java.time.LocalDateTime;
import java.util.*;

/**
 * Created by admin on 2018/6/11.
 */
@Component
public class DataSaveHbaseCommon {
    @Autowired
    private RedisServer rediscConfig ;
    @Autowired
    private HbaseBaseQueryComponent hbaseBaseQueryComponent;
    @Autowired
    private CurrencyInfoHbaseComponent currencyInfoHbaseComponent;
    @Resource
    private CnyToUsdRateComponent cnyToUsdRateComponent;

    @Value("${tradesHistoryAmount.SumAmount}")
    private String SumAmount;
    @Resource
    private PurchaseOrderMapper purchaseOrderMapper;

    @Resource
    private TradeKlineMapper tradeKlineMapper;
    @Autowired
    private AmqpTemplate rabbitTemplatenew;
    /**
     * 历史交易公共方法
     * @param body
     */
    public void MqDataSaveTradesAction(byte[] body) {
        try {
            Long st = System.currentTimeMillis();
            LogTrace.beginTrace("kline-交易历史消费开始");
            if(body == null){
                LogTrace.info("kline-交易历史消费","body","传入内容为空");
                return;
            }
            String params =new String(body, "UTF-8");
            if(StringUtils.isBlank(params)) return;
            JSONObject jsonObject = JSON.parseObject(params);
            String exchange = jsonObject.get("exchange")!=null?jsonObject.get("exchange").toString():"";
            String symbol = jsonObject.get("symbol")!=null?jsonObject.get("symbol").toString():"";
            Integer exchangeId = jsonObject.get("exchangeId")!=null?Integer.valueOf(jsonObject.get("exchangeId").toString()):null;
            List<TradesHistory> tradesList = jsonObject.get("trades")!=null?JSON.parseArray(JSON.toJSONString(jsonObject.get("trades")),TradesHistory.class):null;
            //如果没有数据返回
            if(tradesList == null || tradesList.size() == 0 || StringUtils.isBlank(exchange) || StringUtils.isBlank(symbol) || null == exchangeId){
                LogTrace.error("kline-交易历史消费",new ErrorDataException(exchange+"-"+symbol+"-"+exchangeId+"无数据或者数据错误"));
                return;
            }
            LogTrace.info("kline-交易历史消费","明细",exchangeId+"-"+exchange+"-"+symbol);
            String redisData = rediscConfig.get(RedisKeysUtils.API_SERVICE_TRADES+"@"+exchangeId+"@"+symbol);
            Long lastTimestamp = Long.MIN_VALUE;
            BigDecimal beforeClosePrice = new BigDecimal(0);//记录上一次收盘价格
            LogTrace.info("kline-交易历史消费","接收最后时间",tradesList.get(tradesList.size()-1).getTimestamp().toString());
            //获取redis中最大时间戳，后面去重使用
            if(StringUtils.isNotBlank(redisData)){
                List<TradesHistory> tradesRedisList = JSON.parseArray(redisData,TradesHistory.class);
                if(tradesRedisList != null && tradesRedisList.size() > 0){
                    lastTimestamp = tradesRedisList.get(tradesRedisList.size()-1).getTimestamp();
                    beforeClosePrice = new BigDecimal(tradesRedisList.get(tradesRedisList.size()-1).getPrice().stripTrailingZeros().toPlainString());
                }
            }
            LogTrace.info("kline-交易历史消费","redis最后时间",String.valueOf(lastTimestamp));
            Map<String,List<TradesHistory>> rowKeyMap = new HashMap<>(); //大于50万数据
            Map<String,TradesHistoryCountFive> fiveAmountMap = new HashMap<>();//保存最新的时间数据5分钟
            Map<String,TradesToKLine> kLineMap = new HashMap<>(); //每分钟k线数据
            //从redis中获取最新的k线数据，并且放入KLineMap中
            String redisLastKlineKey = RedisKeysUtils.API_SERVICE_K_ACTION_TRADES+"@"+exchangeId+"@"+symbol;
            String redisLastKlineData = rediscConfig.get(redisLastKlineKey);
            if(StringUtils.isNotBlank(redisLastKlineData)){
                List<TradesToKLine> lastKline = JSON.parseArray(redisLastKlineData,TradesToKLine.class);
                if(lastKline != null && lastKline.size() > 0){
                    for(TradesToKLine lineData : lastKline){
                        Long nowTimestamp = Long.valueOf(lineData.getTimestamp());
                        LocalDateTime timeRedis = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
                        String redisMinutePart = DateUtils.localDateTimeFormatString(timeRedis,DateUtils.format_ymdhm);
                        kLineMap.put(redisMinutePart,lineData);
                    }
                }
            }
            String basePriod = "";
            TradesToKLine tradesToKLine;
            String todayOpenPrice ="";//今日开盘价格
            String redisPartKey = RedisKeysUtils.API_SERVICE_TRADES_PART+"@"+exchangeId+"@"+symbol;
            String redisLastAmount = rediscConfig.get(redisPartKey);
            if(StringUtils.isNotBlank(redisLastAmount)){
                TradesHistoryCountFive redisPart = JSON.parseObject(redisLastAmount,TradesHistoryCountFive.class);
                fiveAmountMap.put(redisPart.getDateTime(),redisPart);
            }
            List<TradesHistory> tempData ;
            //获取汇率
            /*String rmbUsd = rediscConfig.get(RedisKeysUtils.RMB_USD_RATE);
            if(StringUtils.isBlank(rmbUsd)){
                rmbUsd = cnyToUsdRateComponent.getCnyUsdRate();
            }
            BigDecimal rmbUsdToCny = new BigDecimal(rmbUsd);*/
            String rowKey="";
            BigDecimal SumAmountUsd = new BigDecimal(SumAmount);
           // SumAmountUsd = SumAmountUsd.divide(rmbUsdToCny,4,BigDecimal.ROUND_DOWN);
            String timePart;
            String periodRowKey;
            String hbaseBasePriod;
            //将最后一个数据接收盘价格放入redis中，下一次的开盘去此数据
            String lastTradesPriceRedisKey= RedisKeysUtils.K_TRADES_LINE_LAST_PRICE+"@"+exchangeId+"@"+symbol;
            String lastTradesPriceRedisValue ="";
            BigDecimal openPriceOld = BigDecimal.ZERO;
            //循环去重，按照分钟（rowkey）组装map数据，然后循环查hbase，插入hbase
            for(int i =0;i<tradesList.size();i++){
                TradesHistory tradesHistory = tradesList.get(i);
                tempData = new ArrayList<>();
                Long nowTimestamp = tradesHistory.getTimestamp();
                if(lastTimestamp != Long.MIN_VALUE && lastTimestamp > nowTimestamp){
                    tradesList.remove(tradesHistory);
                    i --;
                    continue;
                }
                tradesHistory.setAmount(tradesHistory.getAmount()!=null?tradesHistory.getAmount().abs():BigDecimal.ZERO);
                LocalDateTime time = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
                //得到当前时间处于哪个时间段，（0-11）分别代表0-59
                int minu = time.getMinute()/5;
                timePart = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdh);
                timePart += minu < 10 ? "0"+minu:minu;
                String lastStr = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymds);
                String minutePart = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);//当前时间转成分钟
                //平台币或者不是usd和usdt结尾的交易对转成美金价格
                tradesHistory = this.conversionPrice(exchangeId,symbol,tradesHistory);
                //查询reids是否已经保存当天第一天数据，如果没有去hbase查，如果还没有当前第一条为0点后第一条数据（无论是不是，可能发布时间存在差异）
                String redisKeyPeriod = RedisKeysUtils.API_SERVICE_TRADES_PRIOD+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                basePriod = rediscConfig.get(redisKeyPeriod);
                if(StringUtils.isBlank(basePriod)){
                    periodRowKey = HbaseTableUtils.API_SERVICE_TRADES_ROWKEY+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                    hbaseBasePriod = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                            , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                            , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                            , periodRowKey);
                    if(StringUtils.isBlank(hbaseBasePriod)){//当前数据为第一条，插入到hbase中
                        //tradesHistory.setPriceCny(tradesHistory.getPrice().multiply(rmbUsdToCny).setScale(6,BigDecimal.ROUND_DOWN).toString());
                        hbaseBasePriod = JSON.toJSONString(tradesHistory);
                        currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                                ,periodRowKey
                                , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                                , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                                ,hbaseBasePriod);
                    }
                    todayOpenPrice = hbaseBasePriod;
                    rediscConfig.set(redisKeyPeriod,hbaseBasePriod,90000L);
                }else{
                    todayOpenPrice = basePriod;
                }
                //插入12点后第一条数据，用于k线计算-选币
                if(time.getHour()>=12){
                    String redisKeyPmOpen= RedisKeysUtils.API_SERVICE_TRADES_PM_PRIOD+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                    String pmOpenPrice = rediscConfig.get(redisKeyPmOpen);
                    if(StringUtils.isBlank(pmOpenPrice)){
                        String periodPmRowKey = HbaseTableUtils.API_SERVICE_TRADES_PM_ROWKEY+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                        String hbaseBasePriod2 = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                                , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                                , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                                , periodPmRowKey);
                        if(StringUtils.isBlank(hbaseBasePriod2)){//当前数据为第一条，插入到hbase中
                            hbaseBasePriod2 = JSON.toJSONString(tradesHistory);
                            currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                                    ,periodPmRowKey
                                    , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                                    , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                                    ,hbaseBasePriod2);
                        }
                        rediscConfig.set(redisKeyPmOpen,hbaseBasePriod2,90000L);
                    }
                }
                //分别计算买和卖总量，公式为交易量乘以价格，如果交易超过50万保存到hbase，计算5分钟内交易总和
                BigDecimal buyOrSellCount = tradesHistory.getAmount().multiply(tradesHistory.getNewPriceCny());
                TradesHistoryCountFive partAmount = fiveAmountMap.get(timePart);
                if(partAmount == null){
                    partAmount = new TradesHistoryCountFive();
                    partAmount.setDateTime(timePart);
                    if("buy".equals(tradesHistory.getSide())){
                        partAmount.setBuyAmount(buyOrSellCount);
                        partAmount.setSellAmount(BigDecimal.valueOf(0));
                    }else{
                        partAmount.setBuyAmount(BigDecimal.valueOf(0));
                        partAmount.setSellAmount(buyOrSellCount);
                    }
                    partAmount.setAmount(tradesHistory.getAmount());
                }else{
                    if("buy".equals(tradesHistory.getSide())){
                        partAmount.setBuyAmount(buyOrSellCount.add(partAmount.getBuyAmount()));
                    }else{
                        partAmount.setSellAmount(buyOrSellCount.add(partAmount.getSellAmount()));
                    }
                    partAmount.setAmount(tradesHistory.getAmount().add(partAmount.getAmount()));//统计所有买卖总量
                }
                fiveAmountMap.put(partAmount.getDateTime(),partAmount);
                if(buyOrSellCount.compareTo(SumAmountUsd) >= 0){//大于50万
                    //获取时间戳转成rowKey，规则：+交易所id+@交易对+@201805020225（年月日时分），买卖单数据保存没意义所以只保存最新一条即可
                    rowKey = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);
                    if(rowKeyMap.containsKey(rowKey)){
                        tempData =rowKeyMap.get(rowKey);
                    }
                    tempData.add(tradesHistory);
                    rowKeyMap.put(rowKey,tempData);
                }
                //每分钟k线数据
                tradesToKLine = kLineMap.get(minutePart);
                if(tradesToKLine == null){
                    tradesToKLine = new TradesToKLine();
                    tradesToKLine.setVolume(tradesToKLine.getVolume());
                    //改成从redis获取
                    lastTradesPriceRedisValue = rediscConfig.get(lastTradesPriceRedisKey);
                    //开盘如果从redis获取，开盘价格要与当前价格比较，然后放入最低、最高中
                    if(StringUtils.isNotBlank(lastTradesPriceRedisValue)){
                        openPriceOld = new BigDecimal(lastTradesPriceRedisValue);
                        tradesToKLine.setOpenPrice(lastTradesPriceRedisValue);
                        if(openPriceOld.compareTo(tradesHistory.getPrice()) > 0){
                            tradesToKLine.setHighestPrice(openPriceOld.stripTrailingZeros().toPlainString());
                            tradesToKLine.setLowestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                        }else{
                            tradesToKLine.setHighestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                            tradesToKLine.setLowestPrice(openPriceOld.stripTrailingZeros().toPlainString());
                        }
                    }else{
                        tradesToKLine.setOpenPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                        tradesToKLine.setHighestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                        tradesToKLine.setLowestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                    }
                    tradesToKLine.setClosePrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                    tradesToKLine.setVolume(tradesHistory.getAmount().stripTrailingZeros().toPlainString());
                }else {
                    tradesToKLine.setClosePrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                    //计算最高、最低、总量
                    BigDecimal highestPriceOld = new BigDecimal(tradesToKLine.getHighestPrice());
                    BigDecimal lowestPriceOld = new BigDecimal(tradesToKLine.getLowestPrice());
                    BigDecimal amountOld = new BigDecimal(tradesToKLine.getVolume());
                    if(tradesHistory.getPrice().compareTo(highestPriceOld) > 0){
                        tradesToKLine.setHighestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                    }
                    if(tradesHistory.getPrice().compareTo(lowestPriceOld) < 0){
                        tradesToKLine.setLowestPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                    }
                    tradesToKLine.setVolume(tradesHistory.getAmount().add(amountOld).setScale(2,BigDecimal.ROUND_DOWN).toString());
                }
                tradesToKLine.setPriceCny(tradesHistory.getNewPriceCny().stripTrailingZeros().toPlainString());
                tradesToKLine.setNewPriceUsd(tradesHistory.getNewPriceUsd());
                tradesToKLine.setTimestamp(nowTimestamp.toString());
                tradesToKLine.setPingPrice(tradesHistory.getPingPrice());
                tradesToKLine.setRmbUsdt(tradesToKLine.getRmbUsdt());
                kLineMap.put(minutePart,tradesToKLine);
                //RedisKeysUtils.K_TRADES_LINE_LAST_PRICE+"@"+exchangeId+"@"+symbol将最后一个数据接收盘价格放入redis中，下一次的开盘去此数据
                rediscConfig.set(lastTradesPriceRedisKey,tradesHistory.getPrice());
            }
            if(tradesList == null || tradesList.size() == 0){
                return;
            }
            //测试统计中断原因，值打印交易id-12-btc记录redis中
            /*if(exchangeId == 12 && symbol.equals("BTC/USDT")){
                String rowKeyTemTest = RedisKeysUtils.K_LINE_TEST_TEMP_TEST;
                String temReowkey;
                String temStringdata;
                List<TradesHistory> tempListTEst;
                for(TradesHistory history:tradesList){
                    Long nowTimestamp = history.getTimestamp();
                    LocalDateTime time = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
                    temReowkey = rowKeyTemTest+"@"+DateUtils.localDateTimeFormatString(time,DateUtils.format_ymds)+time.getHour();
                    temStringdata = rediscConfig.get(temReowkey);
                    history.setId(DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm));
                    if(StringUtils.isNotBlank(temStringdata)){
                        tempListTEst = JSON.parseArray(temStringdata,TradesHistory.class);
                        tempListTEst.add(history);
                    }else{
                        tempListTEst = new ArrayList<>();
                        tempListTEst.add(history);
                    }
                    rediscConfig.set(temReowkey,JSON.toJSONString(tempListTEst));
                }
            }*/
            //测试结束，正式环境取消
            //比较当前价格上涨还是下跌 -- beforeClosePrice
            BigDecimal nowPrice = new BigDecimal(tradesList.get(tradesList.size()-1).getPrice().stripTrailingZeros().toPlainString());
            if(tradesList.size() > 1 && beforeClosePrice.compareTo(BigDecimal.ZERO) == 0){
                beforeClosePrice = new BigDecimal(tradesList.get(tradesList.size()-2).getPrice().stripTrailingZeros().toPlainString());
            }
            int changFlag = nowPrice.compareTo(beforeClosePrice);
            Long start = System.currentTimeMillis();
            String hbaseData="";
            //循环rowKeyMap，循环查询每分钟超过50万总量保存到mysql数据的
            PurchaseOrder purchaseOrder = new PurchaseOrder();
            Date date = new Date();
            for(Map.Entry<String,List<TradesHistory>> temMap :rowKeyMap.entrySet()){
                tempData = temMap.getValue();
                if(tempData != null && tempData.size() > 0){
                    for(TradesHistory th : tempData){
                        purchaseOrder = new PurchaseOrder();
                        purchaseOrder.setCount(th.getAmount().setScale(8,BigDecimal.ROUND_DOWN).stripTrailingZeros().toPlainString());
                        purchaseOrder.setCreateTime(date);
                        purchaseOrder.setDealSign(th.getSide().equals("buy")?"0":"1");
                        purchaseOrder.setExchangeId(exchangeId);
                        purchaseOrder.setTransactionPair(symbol);
                        purchaseOrder.setPrice(th.getPrice().stripTrailingZeros().toPlainString());
                        purchaseOrder.setTimeStamp(th.getTimestamp().toString());
                        purchaseOrder.setPriceCny(th.getNewPriceCny().stripTrailingZeros().toPlainString());
                        purchaseOrderMapper.insert(purchaseOrder);
                    }
                }
            }
            //保存5分钟总和数据
            LogTrace.infoKvs("kline-交易历史消费进入实时资金流入","bigin",new Date().toString());
            this.saveFiveTradesDataToHbase(fiveAmountMap,redisPartKey,exchangeId,symbol);
            //保存k线数据到hbase
            this.saveKLineDataToHbase(kLineMap,exchangeId,symbol,todayOpenPrice,redisLastKlineKey,changFlag);
            LogTrace.info("kline-交易历史消费","保存hbase总时间",String.valueOf(System.currentTimeMillis()-start));
            LogTrace.info("kline-交易历史消费","hbase保存","成功"+exchange+"-"+symbol+"-"+exchangeId);
            //保存数据到redis
            LogTrace.info("kline-交易历史消费","最后时间",tradesList.get(tradesList.size()-1).getTimestamp().toString());
            rediscConfig.set(RedisKeysUtils.API_SERVICE_TRADES+"@"+exchangeId+"@"+symbol,JSON.toJSONString(tradesList));
            //放入30条数据用于第一次展示数据
            String redisKey30Tiao = RedisKeysUtils.API_SERVICE_TRADES_ALL_DATA+"@"+exchangeId+"@"+symbol;
            String redisKey30TiaoData = rediscConfig.get(redisKey30Tiao);
            List<TradesHistory> trades30List = new ArrayList<>();
            if(StringUtils.isNotBlank(redisKey30TiaoData)){
                trades30List = JSON.parseArray(redisKey30TiaoData,TradesHistory.class);
            }
            trades30List.addAll(tradesList);
            if(trades30List.size() > 30){
                int cha = trades30List.size() -30;
                if(cha > 0){
                    for(int z =0;z<cha;z++){
                        trades30List.remove(0);
                    }
                }
            }
            rediscConfig.set(redisKey30Tiao,JSON.toJSONString(trades30List));
            LogTrace.info("kline-交易历史消费","redis保存","成功"+exchange+"-"+symbol+"-"+exchangeId+"-"+String.valueOf(System.currentTimeMillis()-start));
            LogTrace.info("kline-交易历史消费","总耗时",String.valueOf(System.currentTimeMillis()-st));
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("kline-交易历史消费error",e);
        }
    }

    /**
     * 平台币或者非usd和usdt结尾的交易对转成美金交易对，最后转成人民币价格
     * @param exchangeId
     * @param symbol
     * @param tradesHistory
     * @return
     */
    private TradesHistory conversionPrice(Integer exchangeId, String symbol, TradesHistory tradesHistory) {
        String [] sy = symbol.split("/");
        //美元获取汇率
        String rmbUsd = rediscConfig.get(RedisKeysUtils.RMB_USD_RATE);
        if(StringUtils.isBlank(rmbUsd)){
            rmbUsd = cnyToUsdRateComponent.getCnyUsdRate();
        }
        BigDecimal rmbUsdToCny = new BigDecimal(rmbUsd);
        String baseCoin = sy[1];
        baseCoin = baseCoin.toUpperCase();
        if(!"CNY".equals(baseCoin)){
            if(!"USD".equals(baseCoin) && !"USDT".equals(baseCoin)){//查找到对应的usdt或者usd价格
                String[] searchCoin = {baseCoin+"/USDT",baseCoin+"/USD"};
                LocalDateTime time = LocalDateTime.now();
                int i = 0;
                String converPrice = "";
                Integer exchangeIdConvert = rediscConfig.get(RedisKeyApp.EXCHANGE_PING_TAI_B_CONVERT+exchangeId,Integer.class);
                if(exchangeIdConvert != null){
                    exchangeId =exchangeIdConvert;
                }
                for(String coin :searchCoin){
                  //  String redisPriceDate = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymds);
                    String changeRedisKey = RedisKeysUtils.API_SERVICE_K_ACTION_ALL_DATA_PRICE+"@"+exchangeId+"@"+coin;
                    String data = rediscConfig.get(changeRedisKey);
                    KActionTotalPo kActionTotalPo = JSON.parseObject(data,KActionTotalPo.class);
                    if(kActionTotalPo != null){
                        converPrice = kActionTotalPo.getLastOpenPrice();
                        break;
                    }
                }
                if(StringUtils.isNotBlank(converPrice)){
                    BigDecimal conPrice = new BigDecimal(converPrice);
                    tradesHistory.setNewPriceCny(tradesHistory.getPrice().multiply(conPrice));
                    tradesHistory.setNewPriceUsd(tradesHistory.getPrice().multiply(conPrice));
                    tradesHistory.setPingPrice(conPrice);
                }
            }else{
                tradesHistory.setNewPriceCny(tradesHistory.getPrice());
                tradesHistory.setNewPriceUsd(tradesHistory.getPrice());
            }
            tradesHistory.setRmbUsdt(rmbUsdToCny);
            tradesHistory.setNewPriceCny(tradesHistory.getNewPriceCny()!=null?tradesHistory.getNewPriceCny().multiply(rmbUsdToCny):tradesHistory.getPrice().multiply(rmbUsdToCny));
        }
        return tradesHistory;
    }

    /**
     * 保存k线数据到hbase中
     * @param kLineMap
     */
    private void saveKLineDataToHbase(Map<String, TradesToKLine> kLineMap,int exchangeId,String symbol,String basePriod,String redisLastKlineKey,int changFlag) throws Exception {
        List<TradesToKLine> kLineList = new ArrayList<>();
        for(Map.Entry<String,TradesToKLine> kline : kLineMap.entrySet()){
            TradesToKLine line = kline.getValue();
            kLineList.add(line);
        }
        Collections.sort(kLineList, new Comparator<TradesToKLine>() {
            @Override
            public int compare(TradesToKLine h1, TradesToKLine h2) {
                Long t1 = Long.valueOf(h1.getTimestamp());
                Long t2 = Long.valueOf(h2.getTimestamp());
                if(t1.longValue() > t2.longValue()){
                    return 1;
                }else if(t1 == t2){
                    return 0;
                }else{
                    return -1;
                }
            }
        });
        String rowKey;
       // String rowKeyAllDataRedis = RedisKeysUtils.API_SERVICE_K_TRADES_ALL_DATA+"@"+exchangeId+"@"+symbol;
        int listSize = kLineList.size();
        for(int i=0;i<listSize;i++){
            TradesToKLine lastLine = kLineList.get(i);
            Long nowTimestamp = Long.valueOf(lastLine.getTimestamp());
            LocalDateTime time = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
            String redisPriceDate = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymds);
            String changeRedisKey = RedisKeysUtils.API_SERVICE_K_ACTION_CHANGE_PRICE+"@"+exchangeId+"@"+symbol+"@"+redisPriceDate;
            String allDataRedisKey = RedisKeysUtils.API_SERVICE_K_ACTION_ALL_DATA_PRICE+"@"+exchangeId+"@"+symbol;
            String oldDate = rediscConfig.get(changeRedisKey);
            //计算最高、最低价格
            KActionTotalPo kAction = new KActionTotalPo();
            if(StringUtils.isNotBlank(oldDate)){
                kAction = JSON.parseObject(oldDate,KActionTotalPo.class);
                BigDecimal lastMaxPrice = new BigDecimal(lastLine.getHighestPrice());
                BigDecimal lastMinPrice = new BigDecimal(lastLine.getLowestPrice());
                BigDecimal lastMaxPriceOld = new BigDecimal(StringUtils.isNotBlank(kAction.getMaxPrice())?kAction.getMaxPrice():"0");
                if(StringUtils.isNotBlank(kAction.getMinPrice())){
                    BigDecimal lastMinPriceOld = new BigDecimal(kAction.getMinPrice());
                    if(lastMinPrice.compareTo(lastMinPriceOld) < 0){
                        kAction.setMinPrice(lastLine.getLowestPrice());
                    }
                }else{
                    kAction.setMinPrice(lastLine.getLowestPrice());
                }
                if(lastMaxPrice.compareTo(lastMaxPriceOld) > 0){
                    kAction.setMaxPrice(lastLine.getHighestPrice());
                }
            }else{
                kAction.setMaxPrice(lastLine.getHighestPrice());
                kAction.setMinPrice(lastLine.getLowestPrice());
            }
            rediscConfig.set(changeRedisKey,JSON.toJSONString(kAction),90000L);
            //存放24h内的rowkey到redis中 最多1440条数据
            String rowKey24H = RedisKeysUtils.TRADES_ROW_KEY_24_H+"@"+exchangeId+"@"+symbol;
            if(i == listSize-1){
                //最后一条计算开盘、收盘，涨幅、价格转人民币放入redis中,先获取，如果此处能获取到替换，否则保留原来的24h涨幅、1h涨幅、今日涨幅
                oldDate = rediscConfig.get(changeRedisKey);
                if(StringUtils.isBlank(oldDate)){//为空查询昨日的
                    redisPriceDate = DateUtils.localDateTimeFormatString(time.minusDays(1),DateUtils.format_ymds);
                    String changeRedisKeyYestoday = RedisKeysUtils.API_SERVICE_K_ACTION_CHANGE_PRICE+"@"+exchangeId+"@"+symbol+"@"+redisPriceDate;
                    oldDate = rediscConfig.get(changeRedisKeyYestoday);
                }
                KActionTotalPo kActionTotalPo;
                if(StringUtils.isNotBlank(oldDate)){
                    kActionTotalPo = JSON.parseObject(oldDate,KActionTotalPo.class);
                }else{
                    kActionTotalPo = new KActionTotalPo();
                    kActionTotalPo.setChangePrice("0");
                    kActionTotalPo.setChangePrice1H("0");
                    kActionTotalPo.setChangePriceToday("0");
                }
              //  TradesToKLine lastLine = kLineList.get(kLineList.size()-1);
                kActionTotalPo.setNewPriceUsd(lastLine.getNewPriceUsd());
                kActionTotalPo.setLastOpenPrice(lastLine.getClosePrice());
                kActionTotalPo.setChangFlag(changFlag);
                //最新交易价格转人民币价格
                BigDecimal closePrice = new BigDecimal(lastLine.getClosePrice());
                //获取汇率
               /* String rmbUsd = rediscConfig.get(RedisKeysUtils.RMB_USD_RATE);
                if(StringUtils.isBlank(rmbUsd)){
                    rmbUsd = cnyToUsdRateComponent.getCnyUsdRate();
                }
                closePrice = closePrice.multiply(new BigDecimal(rmbUsd)).setScale(2,BigDecimal.ROUND_DOWN);*/
                kActionTotalPo.setOpenPriceCny(getConverCut(new BigDecimal(lastLine.getPriceCny())));
                //涨幅计算，线获取今日第一条数据的开盘价格作为基期计算
                /*BigDecimal chang = new BigDecimal(0);
                BigDecimal chang1H = new BigDecimal(0);
                BigDecimal changToday = new BigDecimal(0);*///今日涨幅
                //获取24小时前收盘价格(K) 计算24小时涨幅
                BigDecimal openPrice24H = this.getChange24HOr1H(exchangeId,symbol,lastLine.getTimestamp(),24);
                BigDecimal closePriceNow = new BigDecimal(lastLine.getClosePrice());
                if(null != openPrice24H && openPrice24H.compareTo(BigDecimal.valueOf(-100000))!=0){
                    if(openPrice24H.compareTo(BigDecimal.valueOf(0)) != 0){//（当前分钟收盘-24h开盘）除以1小时收盘,24小时涨幅计算
                        BigDecimal chang = closePriceNow.subtract(openPrice24H).divide(openPrice24H,5,BigDecimal.ROUND_DOWN).multiply(BigDecimal.valueOf(100)).setScale(2,BigDecimal.ROUND_DOWN);
                        kActionTotalPo.setChangePrice(chang.stripTrailingZeros().toPlainString());
                        kActionTotalPo.setChaPriceCny24H(getConverCut(closePriceNow.subtract(openPrice24H)));
                    }
                }
                BigDecimal openPrice1H = this.getChange24HOr1H(exchangeId,symbol,lastLine.getTimestamp(),1);//获取1小时开盘价格
                if(null != openPrice1H && openPrice1H.compareTo(BigDecimal.valueOf(-100000))!=0){//一小时涨幅
                    if(openPrice1H.compareTo(BigDecimal.valueOf(0)) != 0){//（当前分钟收盘-24h收盘）除以24小时收盘,24小时涨幅计算
                        BigDecimal chang1H = closePriceNow.subtract(openPrice1H).divide(openPrice1H,5,BigDecimal.ROUND_DOWN).multiply(BigDecimal.valueOf(100)).setScale(2,BigDecimal.ROUND_DOWN);
                        kActionTotalPo.setChangePrice1H(chang1H.stripTrailingZeros().toPlainString());
                    }
                }
                if(StringUtils.isNotBlank(basePriod)){//计算今日涨幅
                    TradesHistory tradesHistory = JSON.parseObject(basePriod,TradesHistory.class);
                    if(tradesHistory != null && tradesHistory.getPrice() != null){
                       kActionTotalPo.setTodayOpenPrice(tradesHistory.getPrice().stripTrailingZeros().toPlainString());
                       kActionTotalPo.setTodayOpenPriceCny(tradesHistory.getPriceCny());
                       if(closePriceNow.compareTo(BigDecimal.valueOf(-100000))!=0){
                           BigDecimal changToday = closePriceNow.subtract(tradesHistory.getPrice()).divide(closePriceNow,5,BigDecimal.ROUND_DOWN).multiply(BigDecimal.valueOf(100)).setScale(2,BigDecimal.ROUND_DOWN);
                            kActionTotalPo.setChangePriceToday(changToday.stripTrailingZeros().toPlainString());
                        }
                   }
                }
                rediscConfig.set(changeRedisKey,JSON.toJSONString(kActionTotalPo),90000L);
                kActionTotalPo.setTimeLong(Long.valueOf(lastLine.getTimestamp()));
                rediscConfig.set(allDataRedisKey,JSON.toJSONString(kActionTotalPo));

            }else{
                String lastRowKey = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);//当前时间转成分钟
                rowKey = exchangeId+"@"+symbol+"@"+lastRowKey;
                lastLine.setExchangeId(exchangeId);
                lastLine.setPair(symbol);
                currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                        ,rowKey
                        , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                        , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                        ,JSON.toJSONString(lastLine));
                //rowkey插入mysql中
                TradeKline tradeKline = new TradeKline();
                tradeKline.setCreateTime(new Date());
                tradeKline.setRowkey(rowKey);
                tradeKline.setExchangeId(exchangeId);
                tradeKline.setTransactionPairName(symbol);
                tradeKline.setTimeStamp(new  Date(nowTimestamp));
                tradeKline.setTimeVar(lastRowKey);
                tradeKlineMapper.insert(tradeKline);
                //插入mq计算5分钟k-30M等k线数据   发送mq
                rabbitTemplatenew.convertAndSend(ExchangePairAllMqQueue.QUEUE_K_ACTION_EXCHANGE_All,JSON.toJSONString(lastLine));
                //将rowKey放入redis中
                String rowKey24HData = rediscConfig.get(rowKey24H);
                List<String> dataList;
                if(StringUtils.isNotBlank(rowKey24HData)){
                    dataList = JSON.parseArray(rowKey24HData,String.class);
                }else{
                    dataList = new ArrayList<>();
                }
                dataList.add(rowKey);
                if(dataList.size() >1440){//超过1440条去掉第一条数据
                    dataList.remove(0);
                }
                rediscConfig.set(rowKey24H,JSON.toJSONString(dataList));
            }
        }
        List<TradesToKLine> lastKline = new ArrayList<>();
        lastKline.add(kLineList.get(listSize-1));
        /*List<TradesToKLine> lastKline2 = new ArrayList<>();
        if(listSize > 1){
            lastKline2.add(kLineList.get(listSize-2));
        }
        lastKline2.add(kLineList.get(listSize-1));*/
        //最新数据放入缓存中，只放一条用于下次使用
        rediscConfig.set(redisLastKlineKey,JSON.toJSONString(lastKline));
        //最新数据放入缓存中，只放一条用于下次使用,放最后两条数据
        String klineRedis = RedisKeysUtils.API_SERVICE_K_ACTION_TRADES_TWO+"@"+exchangeId+"@"+symbol;
        rediscConfig.set(klineRedis,JSON.toJSONString(kLineList));
    }

    /*public void savePingOrUsdKline(TradesToKLine kLine,int exchangeId,String symbol,LocalDateTime time){
        String lastRowKey = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);//当前时间转成分钟
        String rowKey = exchangeId+"@"+symbol+"@"+lastRowKey;
        currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                ,rowKey
                , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                ,JSON.toJSONString(lastLine));
    }*/

    private BigDecimal getChange24HOr1H(int exchangeId, String symbol, String timestamp,int Hours) throws  Exception{
        BigDecimal openPrice24H = BigDecimal.ZERO;
        //循环获取k线24小时数据知道获取到为止
        String rowKey24H = RedisKeysUtils.TRADES_ROW_KEY_24_H+"@"+exchangeId+"@"+symbol;
        String rowKey24HData = rediscConfig.get(rowKey24H);
        List<String> dataList;
        if(StringUtils.isNotBlank(rowKey24HData)){
            dataList = JSON.parseArray(rowKey24HData,String.class);
        }else{
            return openPrice24H;
        }
        Map<String,String> dataMap = new HashMap<>();
        for(String str : dataList){
            dataMap.put(str,str);
        }
        LocalDateTime time = LocalDateTime.now();
        LocalDateTime timeOld = time.minusMinutes(Hours*60);
       // String strDate24 = "";
        //循环判断是否存在rowkey
        int count = 60*Hours;
        String rowKey = "";
        for(int i=0;i<count;i++){
            String strDate24 = DateUtils.localDateTimeFormatString(timeOld,DateUtils.format_ymdhm);
            rowKey = exchangeId+"@"+symbol+"@"+strDate24;
            if(dataMap.containsKey(rowKey)) {
                LogTrace.info("获取24小时开盘价格", "rowkey", rowKey);
                String kHbaseDate = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                        , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                        , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                        , rowKey);
                String searchRowKey = "";
                if (Hours == 24) {
                    searchRowKey = RedisKeysUtils.K_LINE_CHANGE_24H_SEARCH_ROWKEY + "@" + exchangeId + "@" + symbol;
                } else {
                    searchRowKey = RedisKeysUtils.K_LINE_CHANGE_1H_SEARCH_ROWKEY + "@" + exchangeId + "@" + symbol;
                }
                rediscConfig.set(searchRowKey, rowKey);
                //String openPriceStr;
                String closePriceStr;
                if (StringUtils.isNotBlank(kHbaseDate)) {
                    Object listArray = new JSONTokener(kHbaseDate).nextValue();
                    if (listArray instanceof net.sf.json.JSONArray) {
                        JSONArray jsonArray = JSON.parseArray(kHbaseDate);
                        closePriceStr = jsonArray.getString(4);
                        openPrice24H = new BigDecimal(closePriceStr);
                    } else {
                        JSONObject jsonObject3 = JSON.parseObject(kHbaseDate);
                        TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
                        closePriceStr = tradesToKLine.getClosePrice();
                        openPrice24H = new BigDecimal(closePriceStr);
                    }
                    break;
                }
            }
            timeOld = timeOld.plusMinutes(1);
        }
        return openPrice24H;
    }

    /**
     * 获取24h收盘价格
     * @param exchangeId
     * @param symbol
     * @param timestamp
     * @return
     */
    private BigDecimal getClosePrice24H(int exchangeId, String symbol, String timestamp,int Hours) throws  Exception{
        BigDecimal openPrice24H = BigDecimal.ZERO;
        //循环获取k线24小时数据知道获取到为止
       // Long nowTimestamp = Long.valueOf(timestamp);
       // LocalDateTime time = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
        LocalDateTime time = LocalDateTime.now();
        String nowDate = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);
        BigDecimal nowDouble = new BigDecimal(nowDate);
        LocalDateTime timeOld = time.minusMinutes(Hours*60);
        String strDate24 = DateUtils.localDateTimeFormatString(timeOld,DateUtils.format_ymdhm);
        BigDecimal strDate24Double = new BigDecimal(strDate24);
        while (strDate24Double.compareTo(nowDouble) <=0) {
            String rowKey = exchangeId+"@"+symbol+"@"+strDate24;
            LogTrace.info("获取24小时开盘价格","rowkey",rowKey);
            String kHbaseDate = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                    , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                    , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                    , rowKey);
            String searchRowKey ="";
            if(Hours == 24){
                searchRowKey = RedisKeysUtils.K_LINE_CHANGE_24H_SEARCH_ROWKEY+"@"+exchangeId+"@"+symbol;
            }else{
                searchRowKey = RedisKeysUtils.K_LINE_CHANGE_1H_SEARCH_ROWKEY+"@"+exchangeId+"@"+symbol;
            }
            rediscConfig.set(searchRowKey,rowKey);
            String openPriceStr ;
            String closePriceStr ;
            if(StringUtils.isNotBlank(kHbaseDate)){
                Object listArray = new JSONTokener(kHbaseDate).nextValue();
                if(listArray instanceof  net.sf.json.JSONArray){
                    JSONArray jsonArray = JSON.parseArray(kHbaseDate);
                    openPriceStr = jsonArray.getString(1);
                    closePriceStr = jsonArray.getString(4);
                    openPrice24H = new BigDecimal(closePriceStr);
                }else{
                    JSONObject jsonObject3 = JSON.parseObject(kHbaseDate);
                    TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
                    openPriceStr = tradesToKLine.getOpenPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    openPrice24H = new BigDecimal(closePriceStr);
                }
                break;
                /*TradesToKLine kLine = JSON.parseObject(kHbaseDate,TradesToKLine.class);
                if(kLine != null &&  StringUtils.isNotBlank(kLine.getOpenPrice())){
                    openPrice24H = new BigDecimal(kLine.getOpenPrice());
                    break;
                }*/
            }else{
                openPrice24H = BigDecimal.valueOf(-100000);
                 break;
               /* if(Hours == 24){
                    timeOld = timeOld.plusMinutes(60);
                }else{
                    timeOld = timeOld.plusMinutes(3);
                }
                strDate24 = DateUtils.localDateTimeFormatString(timeOld,DateUtils.format_ymdhm);
                strDate24Double = new BigDecimal(strDate24);*/
            }
        }
        return openPrice24H;
    }

    /**
     * 保存历史交易到habase中
     * @param fiveAmountMap
     */
    private void saveFiveTradesDataToHbase(Map<String, TradesHistoryCountFive> fiveAmountMap,String redisPartKey,int exchangeId,String symbol) throws Exception{
        LogTrace.infoKvs("记录实时资金流入start","exchangeId",exchangeId+"","symbol",symbol,"fiveAmountMap",fiveAmountMap.size()+"","redisPartKey",redisPartKey);
        String rowKey;
        List<TradesHistoryCountFive> allPartList = new ArrayList<>();
        for(Map.Entry<String,TradesHistoryCountFive> fiveAmount : fiveAmountMap.entrySet()){
            TradesHistoryCountFive fiveNow = fiveAmount.getValue();
            allPartList.add(fiveNow);
        }
        //排序->保存
        Collections.sort(allPartList, new Comparator<TradesHistoryCountFive>() {
            @Override
            public int compare(TradesHistoryCountFive h1, TradesHistoryCountFive h2) {
                Long t1 = Long.valueOf(h1.getDateTime());
                Long t2 = Long.valueOf(h2.getDateTime());
                if(t1.longValue() < t2.longValue()){
                    return 1;
                }else if(t1 == t2){
                    return 0;
                }else{
                    return -1;
                }
            }
        });
        List<TradesHistoryCountFive> list24Data;
        LogTrace.info("记录实时资金流入排序结束","allPartList",allPartList.size()+"");
        //循环保存5分钟段落数据到redis获hbase
        for(int i=allPartList.size()-1;i>=0;i--){
            TradesHistoryCountFive historyCountFive = allPartList.get(i);
            if(i == 0){
                rediscConfig.set(redisPartKey,JSON.toJSONString(historyCountFive));
                continue;
            }else{//以前的数据计算总额和买卖差、符号
                BigDecimal bugSuSell = historyCountFive.getBuyAmount().subtract(historyCountFive.getSellAmount());
                if(historyCountFive.getBuyAmount().compareTo(historyCountFive.getSellAmount()) > 0){
                    historyCountFive.setInflow("+");
                }
                historyCountFive.setBuySellAmount(bugSuSell);
                //放入24小时288条数据到redis中
                String redisKey24HData = RedisKeysUtils.API_SERVICE_TRADES_24H_DATA+"@"+exchangeId+"@"+symbol;
                String redis24HData = rediscConfig.get(redisKey24HData);
                //LogTrace.info("记录实时资金流入redis","redis24HData",redis24HData);
                if(StringUtils.isNotBlank(redis24HData)){
                    list24Data = JSON.parseArray(redis24HData,TradesHistoryCountFive.class);
                }else{
                    list24Data = new ArrayList<>();
                }
                //获取最后一次数据
                String historyRowKey = RedisKeysUtils.API_SERVICE_TRADES_LAST_CAPITAL_FLOW_DATA+"@"+exchangeId+"@"+symbol;
                String historyStrData = rediscConfig.get(historyRowKey);
                boolean isAdd = false;
                if(StringUtils.isNotBlank(historyStrData)){
                    TradesHistoryCountFive historyObjData = JSON.parseObject(historyStrData,TradesHistoryCountFive.class);
                    if(historyObjData !=  null){
                        String oldDatetime = historyObjData.getDateTime();
                        String nowDatetime = historyCountFive.getDateTime();
                        if(StringUtils.isNotBlank(oldDatetime)){
                            oldDatetime = oldDatetime.substring(0,8);
                            nowDatetime = nowDatetime.substring(0,8);
                            if(oldDatetime.equals(nowDatetime)){
                                if(historyObjData.getSellAmountNew() == null){
                                    historyObjData.setSellAmountNew(BigDecimal.ZERO);
                                }
                                if(historyObjData.getBuyAmountNew() == null){
                                    historyObjData.setBuyAmountNew(BigDecimal.ZERO);
                                }
                                if(historyObjData.getAmountNew() == null){
                                    historyObjData.setAmountNew(BigDecimal.ZERO);
                                }
                                historyCountFive.setSellAmountNew(historyObjData.getSellAmountNew().add(historyCountFive.getSellAmount()));
                                historyCountFive.setBuyAmountNew(historyObjData.getBuyAmountNew().add(historyCountFive.getBuyAmount()));
                                historyCountFive.setBuySellAmountNew(historyCountFive.getBuyAmountNew().subtract(historyCountFive.getSellAmount()));
                                historyCountFive.setAmountNew(historyObjData.getAmountNew().add(historyCountFive.getAmount()));
                                isAdd = true;
                            }
                        }
                    }
                }
                if(!isAdd){//如果第一次或者新的一天第一次就要重新插入总和
                    historyCountFive.setSellAmountNew(historyCountFive.getSellAmount());
                    historyCountFive.setBuyAmountNew(historyCountFive.getBuyAmount());
                    historyCountFive.setBuySellAmountNew(historyCountFive.getBuySellAmount());
                    historyCountFive.setAmountNew(historyCountFive.getAmount());
                }
                if(historyCountFive.getBuySellAmountNew().compareTo(BigDecimal.valueOf(0)) > 0){
                    historyCountFive.setInflow("+");
                }else{
                    historyCountFive.setInflow("-");
                }
                list24Data.add(historyCountFive);
                if(list24Data.size() > 288){
                    list24Data.remove(0);
                }
                rediscConfig.set(historyRowKey,JSON.toJSONString(historyCountFive));
                LogTrace.info("记录实时资金流入","交易所id-交易对",redisKey24HData+":"+list24Data.size());
                rediscConfig.set(redisKey24HData,JSON.toJSONString(list24Data));
            }
            rowKey = HbaseTableUtils.API_SERVICE_TRADES_PART__ROWKEY+"@"+exchangeId+"@"+symbol+"@"+historyCountFive.getDateTime();
            //保存hbase
            currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                    ,rowKey
                    , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                    , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                    ,JSON.toJSONString(historyCountFive));
        }
    }
    public static String getConverCut(BigDecimal lastPriceB) {
        BigDecimal result = new BigDecimal(0);
        result = lastPriceB!=null?lastPriceB.setScale(10,BigDecimal.ROUND_DOWN):result;
        if(result.compareTo(BigDecimal.valueOf(1)) >=0){
            result = result.setScale(2,BigDecimal.ROUND_DOWN);
        }else{
            BigDecimal divisor = BigDecimal.ONE;
            MathContext mc = new MathContext(5);
            result = result.divide(divisor, mc);
        }
        return result.stripTrailingZeros().toPlainString();
    }
    /*public void MqDataSaveTradesAction(byte[] body) {
        try {
            LogTrace.beginTrace("kline-交易历史消费开始");
            if(body == null){
                LogTrace.info("kline-交易历史消费","body","传入内容为空");
                return;
            }
            String params =new String(body, "UTF-8");
            if(StringUtils.isBlank(params)) return;
            JSONObject jsonObject = JSON.parseObject(params);
            String exchange = jsonObject.get("exchange")!=null?jsonObject.get("exchange").toString():"";
            String symbol = jsonObject.get("symbol")!=null?jsonObject.get("symbol").toString():"";
            Integer exchangeId = jsonObject.get("exchangeId")!=null?Integer.valueOf(jsonObject.get("exchangeId").toString()):null;
            List<TradesHistory> tradesList = jsonObject.get("trades")!=null?JSON.parseArray(JSON.toJSONString(jsonObject.get("trades")),TradesHistory.class):null;
            //如果没有数据返回
            if(tradesList == null || tradesList.size() == 0 || StringUtils.isBlank(exchange) || StringUtils.isBlank(symbol) || null == exchangeId){
                LogTrace.error("kline-交易历史消费",new ErrorDataException(exchange+"-"+symbol+"-"+exchangeId+"无数据或者数据错误"));
                return;
            }
            rediscConfig.set(RedisKeysUtils.API_SERVICE_TRADES_ALL_DATA+"@"+exchangeId+"@"+symbol,JSON.toJSONString(tradesList));
            LogTrace.info("kline-交易历史消费","明细",exchangeId+"-"+exchange+"-"+symbol);
            String redisData = rediscConfig.get(RedisKeysUtils.API_SERVICE_TRADES+"@"+exchangeId+"@"+symbol);
            Long lastTimestamp = Long.MIN_VALUE;
            LogTrace.info("kline-交易历史消费","接收最后时间",tradesList.get(tradesList.size()-1).getTimestamp().toString());
            //获取redis中最大时间戳，后面去重使用
            if(StringUtils.isNotBlank(redisData)){
                List<TradesHistory> tradesRedisList = JSON.parseArray(redisData,TradesHistory.class);
                if(tradesRedisList != null && tradesRedisList.size() > 0){
                    lastTimestamp = tradesRedisList.get(tradesRedisList.size()-1).getTimestamp();
                }
            }
            LogTrace.info("kline-交易历史消费","redis最后时间",String.valueOf(lastTimestamp));
            Map<String,List<TradesHistory>> rowKeyMap = new HashMap<>(); //大于50万数据
            Map<String,TradesHistoryCountFive> fiveAmountMap = new HashMap<>();//保存最新的时间数据从redis获取
            String redisPartKey = RedisKeysUtils.API_SERVICE_TRADES_PART+"@"+exchangeId+"@"+symbol;
            String redisLastAmount = rediscConfig.get(redisPartKey);
            if(StringUtils.isNotBlank(redisLastAmount)){
                TradesHistoryCountFive redisPart = JSON.parseObject(redisLastAmount,TradesHistoryCountFive.class);
                fiveAmountMap.put(redisPart.getDateTime(),redisPart);
            }
            List<TradesHistory> tempData ;
            //获取汇率
            String rmbUsd = rediscConfig.get(RedisKeysUtils.RMB_USD_RATE);
            if(StringUtils.isBlank(rmbUsd)){
                rmbUsd = cnyToUsdRateComponent.getCnyUsdRate();
            }
            String rowKey="";
            BigDecimal SumAmountUsd = new BigDecimal(SumAmount);
            SumAmountUsd = SumAmountUsd.divide(new BigDecimal(rmbUsd),4,BigDecimal.ROUND_DOWN);
            //循环去重，按照分钟（rowkey）组装map数据，然后循环查hbase，插入hbase
            for(int i =0;i<tradesList.size();i++){
                TradesHistory tradesHistory = tradesList.get(i);
                tempData = new ArrayList<>();
                Long nowTimestamp = tradesHistory.getTimestamp();
                if(lastTimestamp != Long.MIN_VALUE && lastTimestamp >= nowTimestamp){
                    tradesList.remove(tradesHistory);
                    i --;
                    continue;
                }
                LocalDateTime time = DateUtils.getDateTimeOfTimestamp(nowTimestamp);
                //得到当前时间处于哪个时间段，（0-11）分别代表0-59
                int minu = time.getMinute()/5;
                String timePart = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdh);
                timePart += minu < 10 ? "0"+minu:minu;
                String lastStr = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymds);
                //查询reids是否已经保存当天第一天数据，如果没有去hbase查，如果还没有当前第一条为0点后第一条数据（无论是不是，可能发布时间存在差异）
                String redisKeyPeriod = RedisKeysUtils.API_SERVICE_TRADES_PRIOD+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                String basePriod = rediscConfig.get(redisKeyPeriod);
                if(StringUtils.isBlank(basePriod)){
                    String periodRowKey = HbaseTableUtils.API_SERVICE_TRADES_ROWKEY+"@"+exchangeId+"@"+symbol+"@"+lastStr;
                    String hbaseBasePriod = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                            , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                            , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                            , periodRowKey);
                    if(StringUtils.isBlank(hbaseBasePriod)){//当前数据为第一条，插入到hbase中
                        hbaseBasePriod = JSON.toJSONString(tradesHistory);
                        currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                                ,periodRowKey
                                , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                                , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                                ,hbaseBasePriod);
                    }
                    rediscConfig.set(redisKeyPeriod,hbaseBasePriod,90000L);
                }
                //分别计算买和卖总量，公式为交易量乘以价格，如果交易超过50万保存到hbase，计算5分钟内交易总和
                BigDecimal buyOrSellCount = tradesHistory.getAmount().multiply(tradesHistory.getPrice());
                TradesHistoryCountFive partAmount = fiveAmountMap.get(timePart);
                if(partAmount == null){
                    partAmount = new TradesHistoryCountFive();
                    partAmount.setDateTime(timePart);
                    if("buy".equals(tradesHistory.getSide())){
                        partAmount.setBuyAmount(buyOrSellCount);
                        partAmount.setSellAmount(BigDecimal.valueOf(0));
                    }else{
                        partAmount.setBuyAmount(BigDecimal.valueOf(0));
                        partAmount.setSellAmount(buyOrSellCount);
                    }
                    partAmount.setAmount(tradesHistory.getAmount());
                }else{
                    if("buy".equals(tradesHistory.getSide())){
                        partAmount.setBuyAmount(buyOrSellCount.add(partAmount.getBuyAmount()));
                    }else{
                        partAmount.setSellAmount(buyOrSellCount.add(partAmount.getSellAmount()));
                    }
                    partAmount.setAmount(tradesHistory.getAmount().add(partAmount.getAmount()));//统计所有买卖总量
                }
                fiveAmountMap.put(partAmount.getDateTime(),partAmount);
                if(buyOrSellCount.compareTo(SumAmountUsd) >= 0){//大于50万
                    //获取时间戳转成rowKey，规则：+交易所id+@交易对+@201805020225（年月日时分），买卖单数据保存没意义所以只保存最新一条即可
                    rowKey = DateUtils.localDateTimeFormatString(time,DateUtils.format_ymdhm);
                    if(rowKeyMap.containsKey(rowKey)){
                        tempData =rowKeyMap.get(rowKey);
                    }
                    tempData.add(tradesHistory);
                    rowKeyMap.put(rowKey,tempData);
                }
            }
            if(tradesList == null || tradesList.size() == 0){
                return;
            }
            Long start = System.currentTimeMillis();
            String hbaseData="";
            //循环rowKeyMap，循环查询每分钟超过50万总量保存到hbase数据
            for(Map.Entry<String,List<TradesHistory>> temMap :rowKeyMap.entrySet()){
                rowKey = temMap.getKey()+"@"+exchangeId+"@"+symbol;
                hbaseData = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                        , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                        , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                        , rowKey);
                LogTrace.info("kline-交易历史消费","查询hbase总时间",String.valueOf(System.currentTimeMillis()-start));
                if(StringUtils.isNotBlank(hbaseData)){
                    tempData = (List<TradesHistory>) JSON.parse(hbaseData);
                    tempData.addAll(temMap.getValue());
                }else{
                    tempData = temMap.getValue();
                }
                //保存hbase
                currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                        ,rowKey
                        , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                        , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                        ,JSON.toJSONString(tempData));
                LogTrace.info("kline-交易历史消费","插入hbase总时间",String.valueOf(System.currentTimeMillis()-start));
            }
            //保存5分钟总和数据
            List<TradesHistoryCountFive> allPartList = new ArrayList<>();
            for(Map.Entry<String,TradesHistoryCountFive> fiveAmount : fiveAmountMap.entrySet()){
                TradesHistoryCountFive fiveNow = fiveAmount.getValue();
                allPartList.add(fiveNow);
            }
            //排序->保存
            Collections.sort(allPartList, new Comparator<TradesHistoryCountFive>() {
                @Override
                public int compare(TradesHistoryCountFive h1, TradesHistoryCountFive h2) {
                    Long t1 = Long.valueOf(h1.getDateTime());
                    Long t2 = Long.valueOf(h2.getDateTime());
                    if(t1.longValue() < t2.longValue()){
                        return 1;
                    }else if(t1 == t2){
                        return 0;
                    }else{
                        return -1;
                    }
                }
            });
            //循环保存5分钟段落数据到redis获hbase
            for(int i=0;i<allPartList.size();i++){
                TradesHistoryCountFive historyCountFive = allPartList.get(i);
                if(i == 0){
                    rediscConfig.set(redisPartKey,JSON.toJSONString(historyCountFive));
                    continue;
                }else{//以前的数据计算总额和买卖差、符号
                    BigDecimal bugSuSell = historyCountFive.getBuyAmount().subtract(historyCountFive.getSellAmount());
                    if(historyCountFive.getBuyAmount().compareTo(historyCountFive.getSellAmount()) > 0){
                        historyCountFive.setInflow("+");
                    }
                    historyCountFive.setBuySellAmount(bugSuSell);
                }
                rowKey = HbaseTableUtils.API_SERVICE_TRADES_PART__ROWKEY+"@"+exchangeId+"@"+symbol+"@"+historyCountFive.getDateTime();
                //保存hbase
                currencyInfoHbaseComponent.insertHbaseString(HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME
                        ,rowKey
                        , HbaseTableUtils.API_SERVICE_TRADES_TABLE_NAME_FAMILY
                        , HbaseTableUtils.API_SERVICE_TRADES_QUALIFIER
                        ,JSON.toJSONString(historyCountFive));
            }
            LogTrace.info("kline-交易历史消费","保存hbase总时间",String.valueOf(System.currentTimeMillis()-start));
            LogTrace.info("kline-交易历史消费","hbase保存","成功"+exchange+"-"+symbol+"-"+exchangeId);
            //保存数据到redis
            LogTrace.info("kline-交易历史消费","最后时间",tradesList.get(tradesList.size()-1).getTimestamp().toString());
            rediscConfig.set(RedisKeysUtils.API_SERVICE_TRADES+"@"+exchangeId+"@"+symbol,JSON.toJSONString(tradesList));
            LogTrace.info("kline-交易历史消费","redis保存","成功"+exchange+"-"+symbol+"-"+exchangeId+"-"+String.valueOf(System.currentTimeMillis()-start));
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("kline-交易历史消费",e);
        }
    }*/

}
